package buaa.qel.rpc.common.codec;

import buaa.qel.rpc.common.enumration.SerializeMethod;
import buaa.qel.rpc.common.message.RpcMessage;
import buaa.qel.rpc.common.serializer.Serializer;
import buaa.qel.rpc.common.serializer.hessian.Hessian1Serializer;
import buaa.qel.rpc.common.serializer.hessian.Hessian2Serializer;
import buaa.qel.rpc.common.serializer.kryo.KryoSerializer;
import buaa.qel.rpc.common.serializer.protostuff.ProtostuffSerializer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
@ChannelHandler.Sharable
public class RpcMessageCodec extends MessageToMessageCodec<ByteBuf, RpcMessage> {
    @Override
    public void encode(ChannelHandlerContext ctx, RpcMessage msg, List<Object> outList) throws Exception {
        ByteBuf tempByteBufOut = ctx.alloc().buffer();
        //ByteBuf tempByteBufOut = ByteBufAllocator.DEFAULT.buffer();
        // 1. 4 bytes for magic number
        tempByteBufOut.writeBytes(new byte[]{'b', 'u', 'a', 'a'});
        // 2. 4 bytes for version(float)
        tempByteBufOut.writeFloat(1.0f);
        log.debug("RpcMessage {} {} {}",msg.getMessageType(),msg.getSequenceId(),msg.getSerializeMethod());
        Serializer serializer = null;
        switch (msg.getSerializeMethod()){
            case SERIALIZER_KRYO:
                tempByteBufOut.writeByte(0);
                tempByteBufOut.writeByte(0xff);
                serializer = new KryoSerializer();
                serializeAndWrite(tempByteBufOut,msg,serializer);
                break;
            case SERIALIZER_HESSIAN1:
                tempByteBufOut.writeByte(1);
                tempByteBufOut.writeByte(0xff);
                serializer = new Hessian1Serializer();
                serializeAndWrite(tempByteBufOut,msg,serializer);
                break;
            case SERIALIZER_HESSIAN2:
                tempByteBufOut.writeByte(2);
                tempByteBufOut.writeByte(0xff);
                serializer = new Hessian2Serializer();
                serializeAndWrite(tempByteBufOut,msg,serializer);
                break;
            case SERIALIZER_PROTOSTUFF:
                tempByteBufOut.writeByte(3);
                tempByteBufOut.writeByte(0xff);
                serializer = new ProtostuffSerializer();
                serializeAndWrite(tempByteBufOut,msg,serializer);
                break;
            default:
                break;
        }
        outList.add(tempByteBufOut);
    }

    @Override
    public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> outList) throws Exception {
        int magicNum = in.readInt();
        float version = in.readFloat();
        log.debug("{}, {}", magicNum, version);
        Serializer deserializer = null;
        RpcMessage message = null;
        byte serializeMethod = in.readByte();
        switch (serializeMethod){
            case 0:
                in.readByte();
                deserializer = new KryoSerializer();
                message = readAndDeserialize(in,deserializer);
                break;
            case 1:
                in.readByte();
                deserializer = new Hessian1Serializer();
                message = readAndDeserialize(in,deserializer);
                break;
            case 2:
                in.readByte();
                deserializer = new Hessian2Serializer();
                message = readAndDeserialize(in,deserializer);
                break;
            case 3:
                in.readByte();
                deserializer = new ProtostuffSerializer();
                message = readAndDeserialize(in,deserializer);
                break;
            default:
                break;
        }
        log.debug("{}", message);
        outList.add(message);
    }

    private void serializeAndWrite(ByteBuf tempByteBufOut,RpcMessage msg,Serializer serializer){
        byte[] bytes = serializer.serialize(msg);
        tempByteBufOut.writeInt(bytes.length);
        tempByteBufOut.writeBytes(bytes);
    }

    private RpcMessage readAndDeserialize(ByteBuf in,Serializer deserializer){
        int length = in.readInt();
        log.debug("readAndDeserialize : {}", length);
        byte[] bytes = new byte[length];
        in.readBytes(bytes, 0, length);
        return (RpcMessage) deserializer.deserialize(bytes,RpcMessage.class);
    }
}
